package com.ek.kademlia.task;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ek.common.Bucket;
import com.ek.common.Context;
import com.ek.common.FutureResponse;
import com.ek.kademlia.protocol.impl.DefaultFindValueRequest;
import com.ek.kademlia.protocol.impl.DefaultFindValueResponse;

public class FindValue extends FutureTask<List<Serializable>> {
    private static final transient Logger log = LoggerFactory.getLogger(FindValue.class);

    public FindValue(final String key, final List<Bucket> buckets, final Context context) {
        super(new Callable<List<Serializable>>() {

            public List<Serializable> call() throws Exception {
                List<FutureResponse<DefaultFindValueResponse>> responses = new LinkedList<FutureResponse<DefaultFindValueResponse>>();
                DefaultFindValueRequest request = new DefaultFindValueRequest();
                request.setNodeID(context.getNodeID());
                request.setMessageID(UUID.randomUUID().toString().getBytes());
                request.setKey(key);
                for (Bucket bucket : buckets) {
                    log.debug("send FIND_VALUE rpc to this bucket:" + bucket);
                    FutureResponse<DefaultFindValueResponse> response = context.getComunicationService().sendMessage(request, new InetSocketAddress(bucket.getAddress(), bucket.getPort()));
                    responses.add(response);
                }
                List<Serializable> result = new LinkedList<Serializable>();
                for (FutureResponse<DefaultFindValueResponse> futureResponse : responses) {
                    DefaultFindValueResponse response = futureResponse.get(2, TimeUnit.SECONDS);
                    if (response != null) {
                        log.debug("received FIND_VALUE rpc response from this address:" + response.getSocketAddress() + "this is the response:" + response);
                        result.add(response.getValue());
                    }
                }
                return result;
            }
        });
    }
}
